{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2018-04-12T23:15:19.871361Z",
     "start_time": "2018-04-12T23:15:17.046551Z"
    },
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Populating the interactive namespace from numpy and matplotlib\n"
     ]
    }
   ],
   "source": [
    "#Initialize and load weather dataframe\n",
    "\n",
    "from pyspark import SparkContext\n",
    "sc = SparkContext(master=\"local[4]\")\n",
    "#sc.version\n",
    "\n",
    "import os\n",
    "import sys\n",
    "\n",
    "from pyspark import SparkContext\n",
    "from pyspark.sql import SQLContext\n",
    "from pyspark.sql.types import Row, StructField, StructType, StringType, IntegerType\n",
    "%pylab inline\n",
    "\n",
    "# Just like using Spark requires having a SparkContext, using SQL requires an SQLContext\n",
    "sqlContext = SQLContext(sc)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2018-04-12T23:15:20.015240Z",
     "start_time": "2018-04-12T23:15:19.873249Z"
    },
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\u001b[34mNY.parquet\u001b[m\u001b[m NY.tgz\r\n"
     ]
    }
   ],
   "source": [
    "from os.path import split,join,exists\n",
    "from os import mkdir,getcwd,remove\n",
    "from glob import glob\n",
    "\n",
    "# create directory if needed\n",
    "\n",
    "\n",
    "data_dir='../../Data'\n",
    "weather_dir=join(data_dir,'Weather')\n",
    "!ls $weather_dir"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2018-04-12T23:15:20.024997Z",
     "start_time": "2018-04-12T23:15:20.018744Z"
    },
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "directory ../../Data/Weather already exists\n"
     ]
    }
   ],
   "source": [
    "if exists(weather_dir):\n",
    "    print('directory',weather_dir,'already exists')\n",
    "else:\n",
    "    print('making',weather_dir)\n",
    "    mkdir(weather_dir)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2018-04-12T23:15:22.364003Z",
     "start_time": "2018-04-12T23:15:20.027666Z"
    },
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "removing ../../Data/Weather/NY.tgz\n",
      "removing ../../Data/Weather/NY.parquet\n",
      "curl https://mas-dse-open.s3.amazonaws.com/Weather/by_state/NY.tgz > ../../Data/Weather/NY.tgz \n",
      "  % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current\n",
      "                                 Dload  Upload   Total   Spent    Left  Speed\n",
      "100 22.1M  100 22.1M    0     0  12.2M      0  0:00:01  0:00:01 --:--:-- 12.2M\n",
      "-rw-r--r--  1 yoavfreund  staff    22M Apr 12 16:15 ../../Data/Weather/NY.tgz\n"
     ]
    }
   ],
   "source": [
    "file_index='NY'\n",
    "zip_file='%s.tgz'%(file_index) #the .csv extension is a mistake, this is a pickle file, not a csv file.\n",
    "old_files='%s/%s*'%(weather_dir,zip_file[:-3])\n",
    "for f in glob(old_files):\n",
    "    print('removing',f)\n",
    "    !rm -rf {f}\n",
    "\n",
    "command=\"curl https://mas-dse-open.s3.amazonaws.com/Weather/by_state/%s > %s/%s \"%(zip_file, weather_dir,zip_file)\n",
    "print(command)\n",
    "!$command\n",
    "!ls -lh $weather_dir/$zip_file"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2018-04-12T23:15:22.912563Z",
     "start_time": "2018-04-12T23:15:22.367798Z"
    },
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "x NY.parquet/\n",
      "x NY.parquet/_SUCCESS\n",
      "x NY.parquet/part-00000-8342bcf4-7fc2-4183-8e11-aefdb4915fbb-c000.snappy.parquet\n",
      "x NY.parquet/part-00001-8342bcf4-7fc2-4183-8e11-aefdb4915fbb-c000.snappy.parquet\n",
      "x NY.parquet/part-00002-8342bcf4-7fc2-4183-8e11-aefdb4915fbb-c000.snappy.parquet\n",
      "x NY.parquet/part-00003-8342bcf4-7fc2-4183-8e11-aefdb4915fbb-c000.snappy.parquet\n",
      "x NY.parquet/part-00004-8342bcf4-7fc2-4183-8e11-aefdb4915fbb-c000.snappy.parquet\n",
      "x NY.parquet/part-00005-8342bcf4-7fc2-4183-8e11-aefdb4915fbb-c000.snappy.parquet\n",
      "x NY.parquet/part-00006-8342bcf4-7fc2-4183-8e11-aefdb4915fbb-c000.snappy.parquet\n",
      "x NY.parquet/part-00007-8342bcf4-7fc2-4183-8e11-aefdb4915fbb-c000.snappy.parquet\n",
      "x NY.parquet/part-00008-8342bcf4-7fc2-4183-8e11-aefdb4915fbb-c000.snappy.parquet\n",
      "x NY.parquet/part-00009-8342bcf4-7fc2-4183-8e11-aefdb4915fbb-c000.snappy.parquet\n",
      "x NY.parquet/part-00010-8342bcf4-7fc2-4183-8e11-aefdb4915fbb-c000.snappy.parquet\n",
      "x NY.parquet/part-00011-8342bcf4-7fc2-4183-8e11-aefdb4915fbb-c000.snappy.parquet\n",
      "x NY.parquet/part-00012-8342bcf4-7fc2-4183-8e11-aefdb4915fbb-c000.snappy.parquet\n",
      "x NY.parquet/part-00013-8342bcf4-7fc2-4183-8e11-aefdb4915fbb-c000.snappy.parquet\n"
     ]
    }
   ],
   "source": [
    "#extracting the parquet file\n",
    "!tar zxvf {weather_dir}/{zip_file} -C {weather_dir}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2018-04-12T23:15:26.363111Z",
     "start_time": "2018-04-12T23:15:22.915423Z"
    },
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "../../Data/Weather/NY.parquet\n",
      "+-----------+-----------+----+--------------------+-----+\n",
      "|    Station|Measurement|Year|              Values|State|\n",
      "+-----------+-----------+----+--------------------+-----+\n",
      "|USC00303452|       PRCP|1903|[00 7E 00 7E 00 7...|   NY|\n",
      "+-----------+-----------+----+--------------------+-----+\n",
      "only showing top 1 row\n",
      "\n"
     ]
    }
   ],
   "source": [
    "weather_parquet = join(weather_dir,zip_file[:-3]+'parquet')\n",
    "print(weather_parquet)\n",
    "df = sqlContext.read.load(weather_parquet)\n",
    "df.show(1)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Comparing GroupBy and ReduceByKey"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2018-04-12T23:15:26.369093Z",
     "start_time": "2018-04-12T23:15:26.365135Z"
    }
   },
   "outputs": [],
   "source": [
    "def Use_Iterator(X):\n",
    "    key,iter=X\n",
    "    s=0\n",
    "    for pair in iter:\n",
    "        s+= pair[1]\n",
    "    return (key,s)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2018-04-12T23:17:13.676544Z",
     "start_time": "2018-04-12T23:17:13.631401Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 1.81 ms, sys: 1.84 ms, total: 3.65 ms\n",
      "Wall time: 41.5 ms\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "A=df.select('Station','Year').rdd.map(lambda row: (row['Station'],row['Year']))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2018-04-12T23:17:22.583157Z",
     "start_time": "2018-04-12T23:17:22.534348Z"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('USC00303452', 1903), ('USC00303452', 1904), ('USC00303452', 1905)]"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "A.take(3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2018-04-12T23:18:19.907611Z",
     "start_time": "2018-04-12T23:18:03.235143Z"
    },
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "84199 8504099\n"
     ]
    }
   ],
   "source": [
    "AA=A\n",
    "for i in range(100):\n",
    "    AA=AA.union(A)\n",
    "print(A.count(),AA.count())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2018-04-12T23:19:04.297893Z",
     "start_time": "2018-04-12T23:19:04.293222Z"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "(404, 4)"
      ]
     },
     "execution_count": 17,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "AA.getNumPartitions(),A.getNumPartitions()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2018-04-12T23:31:09.209234Z",
     "start_time": "2018-04-12T23:31:09.193373Z"
    }
   },
   "outputs": [],
   "source": [
    "AA=AA.repartition(10).cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2018-04-12T23:31:12.022995Z",
     "start_time": "2018-04-12T23:31:12.008661Z"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "(10, 4)"
      ]
     },
     "execution_count": 31,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "AA.getNumPartitions(),A.getNumPartitions()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2018-04-12T23:31:47.799825Z",
     "start_time": "2018-04-12T23:31:17.377838Z"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('USC00307524', 21324534),\n",
       " ('USC00308248', 98745882),\n",
       " ('USC00302610', 138988221)]"
      ]
     },
     "execution_count": 32,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "B=AA.groupBy(lambda pair:pair[0]).map(Use_Iterator)\n",
    "B.take(3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2018-04-12T23:15:27.113284Z",
     "start_time": "2018-04-12T23:15:16.925Z"
    }
   },
   "outputs": [],
   "source": [
    "C=B.collect()\n",
    "C[:4]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2018-04-12T23:32:03.970999Z",
     "start_time": "2018-04-12T23:32:03.953464Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 5.19 ms, sys: 2.28 ms, total: 7.47 ms\n",
      "Wall time: 13.8 ms\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "g=AA.reduceByKey(lambda x,y:x+y)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2018-04-12T23:32:07.413416Z",
     "start_time": "2018-04-12T23:32:04.966996Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 5.78 ms, sys: 3.22 ms, total: 9 ms\n",
      "Wall time: 2.44 s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "C2=g.collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2018-04-12T23:15:27.117519Z",
     "start_time": "2018-04-12T23:15:16.931Z"
    }
   },
   "outputs": [],
   "source": [
    "C2[:4]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2018-04-12T23:15:27.119038Z",
     "start_time": "2018-04-12T23:15:16.933Z"
    }
   },
   "outputs": [],
   "source": [
    "!pip install pyarrow"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2018-04-12T23:15:27.120740Z",
     "start_time": "2018-04-12T23:15:16.934Z"
    }
   },
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import pandas_udf, PandasUDFType"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2018-04-12T23:15:27.121977Z",
     "start_time": "2018-04-12T23:15:16.936Z"
    }
   },
   "outputs": [],
   "source": [
    "@pandas_udf(\"Station string, Measurement string, Year int, State string\", PandasUDFType.GROUPED_MAP)\n",
    "def substract_mean(pdf):\n",
    "    # pdf is a pandas.DataFrame\n",
    "    return len(pdf.columns)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2018-04-12T23:15:27.122989Z",
     "start_time": "2018-04-12T23:15:16.938Z"
    }
   },
   "outputs": [],
   "source": [
    "df.drop('Values').groupby(\"Station\").apply(substract_mean).show(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2018-04-12T23:15:27.124458Z",
     "start_time": "2018-04-12T23:15:16.940Z"
    }
   },
   "outputs": [],
   "source": [
    "df.schema"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2018-04-12T23:15:27.125693Z",
     "start_time": "2018-04-12T23:15:16.944Z"
    }
   },
   "outputs": [],
   "source": [
    "df.show(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2018-04-12T23:15:27.126848Z",
     "start_time": "2018-04-12T23:15:16.947Z"
    }
   },
   "outputs": [],
   "source": [
    "\"Station string, Measurement string, Year int, Values binary, State string\""
   ]
  }
 ],
 "metadata": {
  "anaconda-cloud": {},
  "celltoolbar": "Slideshow",
  "hide_input": false,
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.4"
  },
  "toc": {
   "base_numbering": 1,
   "nav_menu": {
    "height": "263px",
    "width": "252px"
   },
   "number_sections": true,
   "sideBar": true,
   "skip_h1_title": false,
   "title_cell": "Table of Contents",
   "title_sidebar": "Contents",
   "toc_cell": false,
   "toc_position": {},
   "toc_section_display": "block",
   "toc_window_display": false
  },
  "varInspector": {
   "cols": {
    "lenName": 16,
    "lenType": 16,
    "lenVar": 40
   },
   "kernels_config": {
    "python": {
     "delete_cmd_postfix": "",
     "delete_cmd_prefix": "del ",
     "library": "var_list.py",
     "varRefreshCmd": "print(var_dic_list())"
    },
    "r": {
     "delete_cmd_postfix": ") ",
     "delete_cmd_prefix": "rm(",
     "library": "var_list.r",
     "varRefreshCmd": "cat(var_dic_list()) "
    }
   },
   "types_to_exclude": [
    "module",
    "function",
    "builtin_function_or_method",
    "instance",
    "_Feature"
   ],
   "window_display": false
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}
